package cn.fz.rocketmqdemo01.utils;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class RocketmqAdminFactory {
    private final Map<String, MQAdminExt> adminCache = new ConcurrentHashMap<>(16);

    private final Lock lock = new ReentrantLock();

    public MQAdminExt getOrCreateAdmin(String serverAddr) {
        if (adminCache.containsKey(serverAddr)){
            return adminCache.get(serverAddr);
        }
        try {
            lock.lock();
            if (adminCache.containsKey(serverAddr)){
                return adminCache.get(serverAddr);
            }
            DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt(5000L);
            mqAdminExt.setNamesrvAddr(serverAddr);
            mqAdminExt.start();
            adminCache.put(serverAddr, mqAdminExt);
            return mqAdminExt;
        } catch (MQClientException e) {
            throw new RuntimeException("创建admin发生异常",e);
        } finally {
            lock.unlock();
        }
    }
}
